package dm.ass2;

import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import dm.ass1.ProfilingDriver;
import dm.ass2.eval.EvalDriver;
import dm.ass2.io.CodebookItemsWritable;
import dm.ass2.io.CodebookUsersWritable;
import dm.ass2.item.ItemDriver;
import dm.ass2.user.UserDriver;

public class ClusteringDriver extends Configured implements Tool {
	private static final Log LOG = LogFactory.getLog(ClusteringDriver.class);
	
	public static final String OUTPUT_DIR = "output/";
	
	public static final String PROFILE_ST_OUTPUT_USERS = "usersST";
	public static final String PROFILE_SV_OUTPUT_USERS = "usersSV";
	
	public static final String PROFILE_ST_OUTPUT_ITEMS = "itemsST";
	public static final String PROFILE_SV_OUTPUT_ITEMS = "itemsSV";
	
	public static final String CB_ITEMS_FILENAME = "cbitems";
	public static final String CB_USERS_FILENAME = "cbusers";

	private int maxNumberOfIteration = 2;
	private float epsilon = 0.05f;
	private int bestIteration = 0;
	private float bestRMSD = Float.MAX_VALUE;
	
	private Configuration conf;
	private FileSystem fs;
	
	
	
	public ClusteringDriver() {
		conf = new Configuration();
		try {
			fs = FileSystem.get(conf);
		} catch (IOException e) {
			e.printStackTrace();
		}

	}


	@Override
	public int run(String[] args) throws Exception {
		String ST = args[0];
		String SV = args[1];
		
		String numOfItemsClusters = args[2];
		String numOfUsersClusters = args[3];
		
		String[] arguments;
		int ret = 0;
		
		arguments = new String[] {ST, OUTPUT_DIR+PROFILE_ST_OUTPUT_USERS, OUTPUT_DIR+PROFILE_ST_OUTPUT_ITEMS};
		ret = ToolRunner.run(new ProfilingDriver(), arguments);
		if (ret != 0 ) {
			return ret;
		}
		
		arguments = new String[] {SV, OUTPUT_DIR + PROFILE_SV_OUTPUT_USERS, OUTPUT_DIR + PROFILE_SV_OUTPUT_ITEMS};
		ret = ToolRunner.run(new ProfilingDriver(), arguments);
		if (ret != 0 ) {
			return ret;
		}
		
		UserDriver userDriver = new UserDriver();
		userDriver.generateV(ST, numOfItemsClusters, numOfUsersClusters);

		ItemDriver itemDriver = new ItemDriver();
		itemDriver.setNumberOfClusters(Integer.parseInt(numOfItemsClusters), Integer.parseInt(numOfUsersClusters));
		
		EvalDriver evalDriver = new EvalDriver();
		
		float previousRMSD = Float.MAX_VALUE;
		int currIteration = 1;
		boolean stop = false;
		while (currIteration < this.maxNumberOfIteration && !stop) {
			
			arguments = new String[] {OUTPUT_DIR + PROFILE_ST_OUTPUT_USERS};
			ret = ToolRunner.run(userDriver, arguments);
			
			arguments = new String[] {OUTPUT_DIR + PROFILE_ST_OUTPUT_ITEMS};
			ret = ToolRunner.run(itemDriver, arguments);
			
			arguments = new String[] {OUTPUT_DIR + PROFILE_SV_OUTPUT_USERS};
			ret = ToolRunner.run(evalDriver, arguments);
			
			float currRMSD = readCurrEval();
			if (Math.abs(currRMSD - previousRMSD) < epsilon) {
				stop = true;
			}
			
			if (currRMSD < bestRMSD) {
				bestRMSD = currRMSD;
				bestIteration = currIteration;
			}
			
			previousRMSD = currRMSD;
			copyOutput(currIteration);
			
			currIteration++;
		}
		
		makeNiceOutput(bestIteration);
		
		return ret;
	}
	
	
	private void makeNiceOutput(int bestIteration) throws IOException {
		String paddingBestIteration = String.format("%03d", bestIteration);
		
		Path cbItemsPath  = new Path(OUTPUT_DIR + paddingBestIteration + "/" + CB_ITEMS_FILENAME);
		Path cbUsersPath  = new Path(OUTPUT_DIR + paddingBestIteration + "/" + CB_USERS_FILENAME);
		
		List<CodebookItemsWritable> cbItems = readCB(cbItemsPath, CodebookItemsWritable.class);
		List<CodebookUsersWritable> cbUsers = readCB(cbUsersPath, CodebookUsersWritable.class);
		
		printFileTxt(OUTPUT_DIR + "/UserClusters.txt", cbUsers);
		printFileTxt(OUTPUT_DIR + "/ItemClusters.txt", cbItems);
		
		printCodeBook(cbUsers, OUTPUT_DIR + "/CodeBook.txt");
		
	}
	
	private void printCodeBook(List<CodebookUsersWritable> cbUsers, String filename) throws IOException {
		Path outputPath = new Path(filename);

		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}
		
		FSDataOutputStream out = fs.create(outputPath);				

		boolean firstRow = true;
		for (CodebookUsersWritable cbUser : cbUsers) {
			float[] ratings = cbUser.getItemClusterRating();
			if (firstRow) {
				out.writeChars(String.format("%6s", " "));
				for (int i = 0; i < ratings.length; i++) {
					out.writeChars(String.format("%6s", "L" + i));					
				}
				out.writeChar('\n');
			}
			
			boolean firstCol = true;
			for (int i = 0; i < ratings.length; i++) {
				if (firstCol) {
					out.writeChars(String.format("%6s", "K" + cbUser.getUserClusterId()));
					firstCol = false;
				}
				out.writeChars(String.format("%6.2f", ratings[i]));
			}
			firstRow = false;
			out.writeChar('\n');
		}
		out.flush();
		out.close();
	}


	private <T extends Writable> void printFileTxt(String filename, List<T> cbList) throws IOException {
		Path outputPath = new Path(filename);

		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}

		FSDataOutputStream out = fs.create(outputPath);
		for (T cb : cbList) {
			if (cb instanceof CodebookUsersWritable) {
				CodebookUsersWritable cbUser = (CodebookUsersWritable) cb;
				for (int userId : cbUser.getUsrIds()) {
					out.writeChars(String.valueOf(userId));
					out.writeChars(",");
					out.writeChars(String.valueOf(cbUser.getUserClusterId()));
					out.writeChar('\n');
				}
			} else if (cb instanceof CodebookItemsWritable) {
				CodebookItemsWritable cbItem = (CodebookItemsWritable) cb;
				for (int userId : cbItem.getItemIds()) {
					out.writeChars(String.valueOf(userId));
					out.writeChars(",");
					out.writeChars(String.valueOf(cbItem.getItemClusterId()));
					out.writeChar('\n');
				}
			} 
		}

		out.flush();
		out.close();

	}
	
	private <T extends Writable> List<T> readCB (Path p, Class<T> type) throws IOException{
		List<T> cb = new ArrayList<T>();
		FSDataInputStream in = fs.open(p);
		try {
			T cbItem = type.newInstance();
			cbItem.readFields(in);
			while (cbItem != null) {
				cb.add(cbItem);
				cbItem = type.newInstance();
				cbItem.readFields(in);				
			}
		} catch (EOFException e) {
			LOG.info(String.format("Setup: read %d records to cb item.", cb.size()));
		} catch (InstantiationException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		} finally {
			in.close();
		}
		
		return cb;
	}

	private float readCurrEval() {
		try {
			FSDataInputStream in = fs.open(new Path(OUTPUT_DIR + EvalDriver.OUTPUT_REDUCE_DIR + "part-r-00000"));
			
			BufferedReader b = new BufferedReader(new InputStreamReader(in));
			String line = b.readLine();
			String[] keyVal = line.split("\t");
			
			return Float.parseFloat(keyVal[1]);
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		

		return 0;
	}


	private void copyOutput(int currIteration) {
		Path sourceItem = new Path(OUTPUT_DIR + CB_ITEMS_FILENAME);
		Path targetItem = new Path(OUTPUT_DIR + String.format("%03d", currIteration) + "/" + CB_ITEMS_FILENAME);

		Path sourceUser = new Path(OUTPUT_DIR + CB_USERS_FILENAME);
		Path targetUser = new Path(OUTPUT_DIR + String.format("%03d", currIteration) + "/" + CB_USERS_FILENAME);
		
		Path sourceEval = new Path(OUTPUT_DIR + EvalDriver.OUTPUT_REDUCE_DIR + "/" + "part-r-00000");
		Path targetEval = new Path(OUTPUT_DIR + String.format("%03d", currIteration) + "/" + EvalDriver.OUTPUT_FILENAME_NICE);
		
		
		try {
			FileUtil.copy(fs, sourceItem, fs, targetItem, false, true, conf);
			FileUtil.copy(fs, sourceUser, fs, targetUser, false, true, conf);
			FileUtil.copy(fs, sourceEval, fs, targetEval, false, true, conf);
		} catch (IOException e) {
			LOG.error(String.format("unable to copy from %d to %d directory!", sourceItem.getName(), targetItem.getName()));
			e.printStackTrace();
		}		
		
	}


	public static void main(String[] args) throws Exception {
		ClusteringDriver driver = new ClusteringDriver();
		driver.makeNiceOutput(1);
		//driver.copyOutput(1);
		
		//System.out.println(driver.readCurrEval());
		
		//int ret = ToolRunner.run(driver, args);		
		//System.exit(ret);
		
	}
}
